In [1]:
import numpy as np
import networkx as nx
from itertools import chain, combinations, tee
from graph_enumerator import powerset
# For richer sets of operations on graphs, see the graph_enumerator module.
# from graph_enumerator import *
#
In [2]:
def completeDiGraph(nodes):
"""
returns a directed graph with all possible edges
Variables:
nodes are a list of strings that specify the node names
"""
G = nx.DiGraph()
G.add_nodes_from(nodes)
edgelist = list(combinations(nodes,2))
edgelist.extend([(y,x) for x,y in edgelist])
edgelist.extend([(x,x) for x in nodes])
G.add_edges_from(edgelist)
return G
def conditionalSubgraphs(G,condition_list):
"""
Returns a graph generator/iterator such that any conditions specified in condition_list
are met by some subgraph of G.
This is intended to be used in conjunction with completeDiGraph or any graph which subgraphs
are expected to be taken.
Variables:
G is a graph from which subgraphs will be taken.
condition_list is a list of first order functions that define conditions subgraphs of G need to meet to be output.
Functions in condition_list should return a single boolean value for every graph passed into them.
"""
try:
condition_list[0]
except TypeError:
raise TypeError("""
Subsampling from a graph requires passing in a list of conditions encoded
as first-class functions that accept networkX graphs as an input and return boolean values.""")
for edges in powerset(G.edges()):
G_test = G.copy()
G_test.remove_edges_from(edges)
if all([c(G_test) for c in condition_list]):
yield G_test
def create_no_self_loops_condition():
"""
This factory allows us to specify that there are no valid self-loops
This returns a function that takes an graph argument (G).
NB: This is a common assumption of causal graphs, because they are not considered to be extended through time.
"""
def no_self_loops_condition(G):
return not(any([(y,y) in G.edges() for y in G.nodes()]))
return no_self_loops_condition
def create_path_complete_condition(transmit_node_pairs):
"""
This factory allows us to specify that there are valid directed paths between pairs of nodes.
This returns a function that takes an graph argument (G)
and verifies that for the list of node pairs the graph meets those dependency conditions.
NB: This is useful for making known indirect dependencies explicit.
Variables:
node_list is a list of 2-tuples of nodes that will have valid direct paths
from the first of the nodes to the second.
"""
def path_complete_condition(G):
return all([nx.has_path(G,x,y) for x,y in transmit_node_pairs])
return path_complete_condition
def create_no_input_node_condition(node_list):
"""
This factory allows us to specify that no directed can be directed into a set of nodes.
This returns a function that takes an graph argument (G) and verifies that
none of the nodes in node_list are child nodes.
NB: This is useful for making interventions explicit over a set of graphs.
Variables:
node_list is a list of nodes that will have no parents
"""
def no_input_node_condition(G):
return all([G.in_degree(y)==0 for y in node_list])
return no_input_node_condition
def new_conditional_graph_set(graph_set,condition_list):
"""
This returns a copy of the old graph_set and a new graph generator which has
the conditions in condition_list applied to it.
Warning: This function will devour the iterator that you include as the graph_set input,
you need to redeclare the variable as one of the return values of the function.
Thus a correct use would be:
a,b = new_conditional_graph_set(a,c)
The following would not be a correct use:
x,y = new_conditional_graph_set(a,c)
Variables:
graph_set is a graph-set generator
condition_list is a list of first order functions returning boolean values when passed a graph.
"""
try:
condition_list[0]
except TypeError:
raise TypeError("""
Subsampling from a graph requires passing in a list of conditions encoded
as first-class functions that accept networkX graphs as an input and return boolean values.""")
graph_set_newer, graph_set_test = tee(graph_set,2)
def gen():
for G in graph_set_test:
G_test = G.copy()
if all([c(G_test) for c in condition_list]):
yield G_test
return graph_set_newer, gen()
In [3]:
nodes=["a","b","c","d"]
known_ancestral_relations = [("a","b"),("a","c"),("a","d")]
orphan_nodes = ["a"]
f1 = create_path_complete_condition(known_ancestral_relations)
f2 = create_no_input_node_condition(orphan_nodes)
f3 = create_no_self_loops_condition()
G = completeDiGraph(nodes)
graph_set = conditionalSubgraphs(G, [f1, f2,f3])
print(len([graph.edges() for graph in graph_set]))
In [4]:
"""This shows an example of updating a graph with a new condtion.
"""
nodes=["a","b","c","d"]
known_ancestral_relations = [("a","b"),("a","c"),("a","d")]
orphan_nodes = ["a"]
known_ancestral_relations2 = [("b","c")]
f1 = create_path_complete_condition(known_ancestral_relations)
f2 = create_no_input_node_condition(orphan_nodes)
f3 = create_no_self_loops_condition()
f1_2 = create_path_complete_condition(known_ancestral_relations2)
G = completeDiGraph(nodes)
graph_set = conditionalSubgraphs(G, [f1, f2,f3])
graph_set, graph_set2 = new_conditional_graph_set(graph_set,[f1_2])
print(len([graph.edges() for graph in graph_set2]))
print(len([graph.edges() for graph in graph_set]))
In [5]:
"""Note: Your conditions can contradict one another to the point of generating no graphs.
"""
nodes=["a","b","c","d"]
known_ancestal_relations = [("a","b"),("a","c"),("a","d")]
orphan_nodes=["a","c"]
f1 = create_path_complete_condition(known_ancestal_relations)
f2 = create_no_input_node_condition(orphan_nodes)
G = completeDiGraph(nodes)
graph_set = conditionalSubgraphs(G, [f1, f2])
print(len([graph.edges() for graph in graph_set]))
In [6]:
def sample_from_graph(G,func_dictionary=None,k = 1):
"""
This is the function that samples from the rich networkX Bayes Net graph using the parameterization specified in the node attributes.
Variables:
G is the graph being sampled from.
k is the number of samples.
"""
if func_dictionary == None:
func_dictionary = {"choice": np.random.choice}
nodes_dict = G.nodes(data = True)
node_ids = np.array(G.nodes())
state_spaces = [(node[0],node[1]["state_space"]) for node in nodes_dict]
orphans = [node for node in nodes_dict if node[1]["parents"]==[]]
sample_values = np.empty([len(state_spaces),k],dtype='U20')
sampled_nodes = []
for node in orphans:
## sample k values for all orphan nodes
samp_func = string_to_sample_function(node[1]["sample_function"],func_dictionary)
samp_states = node[1]["state_space"]
samp_distribution = node[1]["distribution"]
samp_index = G.nodes().index(node[0])
sample_values[samp_index,:] = samp_func(samp_states,size=[1,k],p=samp_distribution)
sampled_nodes.append(node[0])
while set(sampled_nodes) < set(G.nodes()):
nodes_to_sample = check_if_parents_filled(G,sampled_nodes)
#nodes_to_sample returns a list of node names that need to be sampled
for n in nodes_to_sample:
#extracts the indices of the parents of the node to be sampled and their values
parent_indices = [(parent,G.nodes().index(parent)) for parent in G.node[n]["parents"]]
parent_vals = [(parent[0],sample_values[parent[1],:]) for parent in parent_indices]
#extracts sample index
samp_index = G.nodes().index(n)
sample_values[samp_index,:] = conditional_sampling(G,n,parent_vals,func_dictionary,k)
sampled_nodes.append(n)
return sample_values
def check_if_parents_filled(G,sampled_nodes):
"""
This function will return those nodes who have not yet been sampled, whose parents have been sampled.
Variables:
G is a networkX graph
sampled_nodes are a list of node names
"""
check_nodes = [x for x in G.nodes() if x not in sampled_nodes]
nodes_to_be_sampled = []
for node in G.nodes(data = True):
if (node[0] in check_nodes) & (node[1]["parents"]<=sampled_nodes):
nodes_to_be_sampled.append(node[0])
if len(nodes_to_be_sampled)==0:
raise RuntimeError("You should never be running this when no values are returned")
return nodes_to_be_sampled
def nodeset_query(G,node_set,attrib=[]):
"""
This is a helper function for querying particular attributes from a node
Variables:
G is a networkX style graph
node_set is a list of node names that are in G
attrib are a list of attributes associated with the nodes in G
"""
if len(attrib)==0:
return [node for node in G.nodes(data = True) if node[0] in node_set]
else:
return_val = []
for node in G.nodes(data=True):
if node[0] in node_set:
return_val.append((node[0],{attr:node[1][attr] for attr in attrib}))
return return_val
def conditional_sampling(G,node,parent_vals,func_dictionary, k = 1):
"""
This function takes a graph as input, a node to sample from in that graph and a set of values for the parents of that node.
This function should not be consulted for variables without any parents.
Variables:
G is a networkX style graph
node is a node in G
parent_vals are the values of the parents of node realized k times
returns an array of values
"""
try: node in G
except KeyError:
print("{} is not in graph".format(n))
output = np.empty(k,dtype="U20")
for i in np.arange(k):
par_val_list = []
for parent in parent_vals:
par_val_list.append(tuple([parent[0],parent[1][i]]))
samp_distribution = G.node[node]["distribution"][tuple(par_val_list)]
samp_func = string_to_sample_function(G.node[node]["sample_function"],func_dictionary)
samp_states = G.node[node]["state_space"]
# output.append(samp_func(samp_states,size=[1],p=samp_distribution))
temp_output = samp_func(samp_states,size=1,p=samp_distribution)
output[i] = temp_output[0]
return output
def string_to_sample_function(func_name, func_dictionary=None):
"""
This allows the function to be passed in as a string that is mapped to a first-class function to other methods.
sample_function is a string that maps onto a function in the dictionary defined below.
This takes two arguments a func_dictionary
"""
if func_dictionary == None:
func_dictionary = {"choice": np.random.choice}
try: func_dictionary[func_name]
except KeyError:
print("{} is not a function defined in the dictionary you passed.".format(func_name))
return func_dictionary[func_name]
In [8]:
dict_of_understanding = {"choice": np.random.choice}
edge_list = [
("sprinkler","grass_wet"),
("rain","sprinkler"),
("rain","grass_wet")
]
node_prop_list = [
("rain",
{"state_space":("raining","dry"), "sample_function": "choice",
"parents":[],
"distribution":[.2,.8]
}),
("sprinkler",
{"state_space":("on","off"),"sample_function": "choice",
"parents":["rain"],
"distribution":{(("rain","raining"),):[.01,.99],
(("rain","dry"),):[.4,.6]}
}),
("grass_wet",
{"state_space":("wet","notWet"),"sample_function": "choice",
"parents":["rain","sprinkler"],
"distribution":{(("rain","raining"),("sprinkler","on")):[.99,.01],
(("rain","raining"),("sprinkler","off")):[.8,.2],
(("rain","dry"),("sprinkler","on")):[.9,.1],
(("rain","dry"),("sprinkler","off")):[0,1]}
})
]
func_dictionary = dict_of_understanding
k = 10
G = nx.DiGraph()
G.clear()
G.add_edges_from(edge_list)
G.add_nodes_from(node_prop_list)
test = sample_from_graph(G,func_dictionary,k)
print(test)
print("\n")
# print(json_graph.adjacency_data(G))
In [20]:
def sample_from_graph(G,f_dict=None,k = 1):
if f_dict == None:
f_dict = {"choice": np.random.choice}
n_dict = G.nodes(data = True)
n_ids = np.array(G.nodes())
n_states = [(n[0],n[1]["state_space"]) for n in n_dict]
orphans = [n for n in n_dict if n[1]["parents"]==[]]
s_values = np.empty([len(n_states),k],dtype='U20')
s_nodes = []
for n in orphans:
samp_f = str_to_f(n[1]["sample_function"],
f_dict)
s_states = n[1]["state_space"]
s_dist = n[1]["dist"]
s_idx = G.nodes().index(n[0])
s_values[s_idx,:] = samp_f(s_states,
size=[1,k],p=s_dist)
s_nodes.append(n[0])
while set(s_nodes) < set(G.nodes()):
nodes_to_sample = has_full_parents(G,s_nodes)
for n in nodes_to_sample:
par_indices = [(par,G.nodes().index(par))
for par in G.node[n]["parents"]]
par_vals = [(par[0],s_values[par[1],:])
for par in par_indices]
samp_index = G.nodes().index(n)
s_values[samp_index,:] = cond_samp(G,n,
par_vals,f_dict,k)
s_nodes.append(n)
return s_values
def has_full_parents(G,s_n):
check_n = [x for x in G.nodes() if x not in s_n]
nodes_to_be_sampled = []
for n in G.nodes(data = True):
if (n[0] in check_n) & (n[1]["parents"]<=s_n):
nodes_to_be_sampled.append(n[0])
if len(nodes_to_be_sampled)==0:
raise RuntimeError("A node must be sampled")
return nodes_to_be_sampled
def nodeset_query(G,n_set,n_atr=[]):
if len(n_atr)==0:
return [n for n in G.nodes(data = True)
if n[0] in n_set]
else:
return_val = []
for n in G.nodes(data=True):
if n[0] in node_set:
return_val.append((n[0],
{attr:n[1][attr] for attr in n_atr}))
return return_val
def cond_samp(G,n,par_vals,f_dict, k = 1):
try: n in G
except KeyError:
print("{} is not in graph".format(n))
output = np.empty(k,dtype="U20")
for i in np.arange(k):
val_list = []
for p in par_vals:
val_list.append(tuple([p[0],p[1][i]]))
samp_dist = G.node[n]["dist"][tuple(val_list)]
samp_f = str_to_f(
G.node[n]["sample_function"],f_dict)
samp_states = G.node[n]["state_space"]
temp_output = samp_f(samp_states,
size=1,p=samp_dist)
output[i] = temp_output[0]
return output
def str_to_f(f_name, f_dict=None):
if f_dict == None:
f_dict = {"choice": np.random.choice}
try: f_dict[f_name]
except KeyError:
print("{} is not defined.".format(f_name))
return f_dict[f_name]
In [30]:
dict_of_understanding = {"choice": np.random.choice}
edge_list = [
("sprinkler","grass_wet"),
("rain","sprinkler"),
("rain","grass_wet")
]
node_prop_list = [
("rain",{
"state_space":("yes","no"),
"sample_function": "choice",
"parents":[],
"dist":[.2,.8]}),
("sprinkler",{
"state_space":("on","off"),
"sample_function": "choice",
"parents":["rain"],
"dist":{(("rain","yes"),):[.01,.99],
(("rain","no"),):[.4,.6]}}),
("grass_wet",{
"state_space":("wet","dry"),
"sample_function": "choice",
"parents":["rain","sprinkler"],
"dist":{
(("rain","yes"),("sprinkler","on")):[.99,.01],
(("rain","yes"),("sprinkler","off")):[.8,.2],
(("rain","no"),("sprinkler","on")):[.9,.1],
(("rain","no"),("sprinkler","off")):[0,1]}})]
func_dictionary = dict_of_understanding
k = 10
G = nx.DiGraph()
G.clear()
G.add_edges_from(edge_list)
G.add_nodes_from(node_prop_list)
test = sample_from_graph(G,k=10)
# test = sample_from_graph(G,func_dictionary,k)
print(test)
print("\n")
# print(json_graph.adjacency_data(G))
In [29]:
from networkx.readwrite import json_graph
import datetime as datetime
import json
def json_filename_utility(inputName, file_format = ".json"):
if file_format[0]!=".":
return str(inputName)+"_{}".format(datetime.datetime.today().strftime("%Y_%V_%u_%H%M")) + "." + str(file_format)
return str(inputName)+"_{}".format(datetime.datetime.today().strftime("%Y_%V_%u_%H%M")) + str(file_format)
def json_adj_save(graph,out_file_name=""):
if len(out_file_name)==0:
out_file_name = json_filename_utility("graph_data")
data = json_graph.adjacency_data(graph)
# test_data = json.dumps(data)
with open(out_file_name,"w") as outfile:
json.dump(data, outfile, ensure_ascii=False)
return
def clean_json_adj_load(file_name):
with open(file_name) as d:
json_data = json.load(d)
H = json_graph.adjacency_graph(json_data)
for edge_here in H.edges():
del(H[edge_here[0]][edge_here[1]]["id"])
return H
def clean_json_adj_loads(json_str):
json_data = json.loads(json_str)
H = json_graph.adjacency_graph(json_data)
for edge_here in H.edges():
del(H[edge_here[0]][edge_here[1]]["id"])
return H
# new_data = json.dumps(json_graph.adjacency_data(test_graph), ensure_ascii=False)
In [156]:
x = str(json_graph.adjacency_data(G))
y = ast.literal_eval(x)
In [157]:
clean_json_adj_loads(x)
{'adjacency': [[],
[{'id': 'grass_wet'}, {'id': 'sprinkler'}],
[{'id': 'grass_wet'}]],
'directed': True,
'graph': [],
'multigraph': False,
'nodes': [{'distribution': {(('rain', 'dry'), ('sprinkler', 'off')): [0, 1],
(('rain', 'dry'), ('sprinkler', 'on')): [0.9, 0.1],
(('rain', 'raining'), ('sprinkler', 'off')): [0.8, 0.2],
(('rain', 'raining'), ('sprinkler', 'on')): [0.99, 0.01]},
'id': 'grass_wet',
'parents': ['rain', 'sprinkler'],
'sample_function': 'choice',
'state_space': ('wet', 'notWet')},
{'distribution': [0.2, 0.8],
'id': 'rain',
'parents': [],
'sample_function': 'choice',
'state_space': ('raining', 'dry')},
{'distribution': {(('rain', 'dry'),): [0.4, 0.6],
(('rain', 'raining'),): [0.01, 0.99]},
'id': 'sprinkler',
'parents': ['rain'],
'sample_function': 'choice',
'state_space': ('on', 'off')}]}
In [ ]:
In [144]:
# json_adj_save(G)
data = json_graph.adjacency_data(G)
def check_type_of_keys(dictionary_passed):
newdict = dict()
for key,value in dictionary_passed.items():
if type(key) is tuple:
newdict[str(key)] = value
newdict[str(key)+"_converted"]=True
else:
# print(type(key),"key",key)
newdict[key]=value
newdict[str(key)+"_converted"]=False
if type(value) is list:
for element in value:
# print(type(element),element," element")
if type(element) is dict:
newdict[key] = check_type_of_keys(element)
# print("yup")
# print(type(value),value, "value")
if type(value) is dict:
newdict[key] = check_type_of_keys(value)
# print("yes")
# print(type(value),value)
return newdict
# print(data)
# newdict=dict()
# for key,value in data.items():
# if type(key) is tuple:
# newdict[str(key)]= value
# newdict[str(key)+"_converted"]=True
# else:
# newdict[key]=value
# newdict[str(key)+"_converted"]=False
# if type(value) is dict:
# for key2,value2 in value.items():
# if type(key2) is tuple:
# newdict[str(key2)]= value2
# newdict[str(key2)+"_converted"]=True
# elif type(value2) is dict:
# for key3,value3 in value2.items():
# if type(key3) is tuple:
# newdict[str(key3)]= value3
# newdict[str(key3)+"_converted"]=True
# elif: type(
# else:
# newdict[key3]=value3
# newdict[str(key3)+"_converted"]=False
# else:
# newdict[key2]=value
# newdict[str(key2)+"_converted"]=False
# else:
# newdict[key]=value
# newdict[str(key)+"_converted"]=False
# print(check_type_of_keys(data))
with open("test_file3","w") as outfile:
json.dump(check_type_of_keys(data), outfile, ensure_ascii=False)
# with open("test_file2","w") as outfile:
# json.dump(data, outfile, ensure_ascii=False)
In [145]:
import ast
def undo_conversion(input_dict):
# print([(x,input_dict[x]) for x in input_dict if x.endswith("_converted")])
not_converted_keys = [x for x in input_dict if x.endswith("_converted") and not input_dict[x]]
for key in not_converted_keys:
input_dict.pop(key)
# print([(x,input_dict[x]) for x in input_dict if x.endswith("_converted")])
converted_keys = [x for x in input_dict if x.endswith("_converted") and input_dict[x]]
# print([(x,input_dict[x]) for x in input_dict if x.endswith("_converted")])
for key in converted_keys:
print(input_dict)
holding_val = input_dict.pop(key)
print(key)
print(input_dict)
second_key = strip_end(key,"_converted")
print("\n\n"+second_key+"\n\n")
input_dict[ast.literal_eval(second_key)]= holding_val
for key, value in input_dict.items():
if type(value) is dict:
input_dict[key] = undo_conversion(value)
return input_dict
def strip_end(text, suffix):
# thanks to http://stackoverflow.com/a/1038999/1816995
# print(text)
if not text.endswith(suffix):
# print(text)
return text
# print(text[:len(text)-len(suffix)])
return text[:len(text)-len(suffix)]
with open("test_file1","r") as outfile:
x = json.load(outfile)
x2 = undo_conversion(x)
# json_graph.adjacency_graph(x)
x,x2
Out[145]:
In [87]:
clean_json_adj_load("test_file1")
In [69]:
for key,value in data["nodes"][1]["distribution"].items():
print(key)
In [72]:
str((1,2))
Out[72]:
In [32]:
"""Next Steps (part 1)— Build a hybrid discrete continuous system
"""
edge_list = [
("sprinkler","grass_wet", {"state_space_map": (("on","off"),(0.0,))}),
("rain","sprinkler", {"state_space_map": ((0.0,),("on","off"))}),
("rain","grass_wet", {"state_space_map":((0.0,),(0.0,))})
]
node_prop_list = [
("rain",
{"state_space":(0.0,), "sample_function": "",#function with positive real number output
"parents":[],
"distribution": {
# insert distribution with real number output value
}
}),
("sprinkler",
{"state_space":("on","off"),"sample_function": ""#discrete output,
"parents":["rain"],
"distribution":{
# Insert distribution with real number input and discrete choice output
}
}),
("grass_wet",
{"state_space":(0.0,),"sample_function": "",#real number function type
"parents":["rain","sprinkler"],
"distribution":{
#Insert distribution with real number and discrete choice input and real number output
}
})
]
Next Steps(part 2):
One of the problems with the above approach to sampling is that there is an implicit trial structure that the sampling relies on to govern its steps. This is embedded every time you see a for loop that iterates through k samples.
This is fine for discrete time steps where you have a sample from all nodes for each of those time steps, but when generalizing to continuous time, where there is no predefined trial structure.